package cn.jly.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

/**
 * @author lanyangji
 * @date 2019/12/4 10:50
 */
object SparkStreaming03_QueueStream {

  def main(args: Array[String]): Unit = {

    // 基于内存队列创建DStream，这种方式了解就行
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming03_QueueStream")
    val streamingContext = new StreamingContext(sparkConf, Seconds(4))

    // 创建RDD队列
    val rddQueue: mutable.Queue[RDD[Int]] = new mutable.Queue[RDD[Int]]()

    // 创建QueueInputDStream
    val inputStream: InputDStream[Int] = streamingContext.queueStream(rddQueue, oneAtATime = false)

    // 处理
    inputStream.map((_, 1)).reduceByKey(_ + _).print()

    // 启动任务
    streamingContext.start()

    // 向队列添加数据
    for (elem <- 1 to 5) {
      rddQueue += streamingContext.sparkContext.makeRDD(1 to 300, 10)
      Thread.sleep(2000)
    }

    streamingContext.awaitTermination()
  }
}
